/**
 * org.vz.product.collect.maintain.service.impl
 * @since : jdk 1.8
 */
package org.vz.product.collect.maintain.service.impl;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import javax.jms.JMSException;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.BrokerInfo;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.stereotype.Service;
import org.vz.product.collect.maintain.config.AcqclusterConfig;
import org.vz.product.collect.maintain.domain.ActivemqHost;
import org.vz.product.collect.maintain.service.AcqclusterService;
import org.vz.product.collect.maintain.util.LogUtil;
import org.vz.product.collect.maintain.util.StringUtil;

/**
 * @author      chenhuaijin
 * @CreateTime  2017年9月14日 上午11:27:09
 * @version     1.0.0
 * @description TODO
 */
@Service
public class AcqclusterServiceImpl implements AcqclusterService{
	
	@Autowired
	AcqclusterConfig ac;
	@Autowired
	JmsMessagingTemplate jmsMessagingTemplate;
	private LogUtil log = new LogUtil(getClass());
	
	public static String previousHostIp = null;//service没被使用过
	
	private static final int SESSION_TIMEOUT = 5000;//ms
	
	
	/* (non-Javadoc)
	 * @see org.vz.product.collect.maintain.service.AcqclusterService#currentState()
	 */
	@Override
	public List<ActivemqHost> currentState()  {
		Map<String,String> map  = null;
		String pcName = null;
		ActiveMQConnectionFactory conn =  (ActiveMQConnectionFactory)(jmsMessagingTemplate.getConnectionFactory());
	   	ActiveMQConnection an = null;
	
		try {
			an = (ActiveMQConnection)(conn.createConnection());
			an.start();
			BrokerInfo bi =	an.getBrokerInfo();
			String brokerUrl = bi.getBrokerURL();//得到当前连接的集群的url形如：tcp://pcName:port
			an.stop();
			an.close();
			pcName = StringUtil.substringBetween(brokerUrl, "//",":");
			
			map= ac.mapNameIp();
		} catch (JMSException e1) {
			
			e1.printStackTrace();
			log.error("activemq集群全部不可使用，信息为:", e1);
			
			List<ActivemqHost> rlt = allTheSame(true, pcName, map);
			
			return rlt;
		}

		
		String currentHostIp = map.get(pcName);
			
		if(previousHostIp == null){//service没被使用过
			if(pcName != null){
				if(pcName.equals(AcqclusterConfig.previousHostName)){//在系统启动这时，一定要保证三个集群节点都能用
					List<ActivemqHost> rlt = allTheSame(false, pcName, map);
					previousHostIp = currentHostIp;
					return rlt;
				}else{
						List<ActivemqHost> list = knowResult(map,pcName,null);
						previousHostIp = currentHostIp;
						return list;					
					}
			}else{
				List<ActivemqHost> rlt = allTheSame(true, pcName, map);
				previousHostIp = currentHostIp;
				return rlt;
			}
			
		}else{//使用过
			if(previousHostIp.equals(currentHostIp)){
				List<ActivemqHost> list = knowResult(map,pcName,null);
				previousHostIp = currentHostIp;
				return list;
			}else{//上一次的可能down
				List<ActivemqHost> list = knowResult(map,pcName,previousHostIp);
				previousHostIp = currentHostIp;
				return list;

			}
		}
	}
	
    /**
     * 连接zookeeper server
     */
	private ZooKeeper connect() {
    	ZooKeeper zk = null;
    	String hosts= ac.ruleString();
        try {
			zk = new ZooKeeper(hosts, SESSION_TIMEOUT, new Watcher(){

				@Override
				public void process(WatchedEvent arg0) {
					
				}});
		} catch (IOException e) {
			e.printStackTrace();
			log.error("zookeeper连接出错，信息为:", e);
			return null;
		}
        return zk;
    }
	/**
	 * 
	 * @description  返回当前的集群的使用情况
	 * @author       chenhuaijin
	 * @CreateTime   2017年9月14日 下午6:33:40
	 * @param map 所有主机名与Ip的映射
	 * @param currentHostName 当前master的主机名
	 * @param previousHostIp 上一次的master的ip
	 * @return
	 */
	private List<ActivemqHost> knowResult(Map<String,String> map,String currentHostName,String previousHostIp){
		List<String> list = null;
		ZooKeeper zk = null;
		
		zk = connect();
		if(zk !=null ){
			try {
				list= zk.getChildren("/activemq/leveldb-stores", true);
				zk.close();
			} catch (KeeperException | InterruptedException e1) {
				e1.printStackTrace();
				log.error("zookeeper连接出错，信息为:", e1);
				
				ActivemqHost ah = null;
				List<ActivemqHost> rlt = new ArrayList<ActivemqHost>();
				for(Entry<String, String> e:map.entrySet()){
					ah = new ActivemqHost();
					ah.setName(e.getKey());
					ah.setIp(e.getValue());
					ah.setCurrentUse(false);
					ah.setBroken(true);
					rlt.add(ah);							
				}
				return rlt;
			}
			if(list.size() == 3){//有台down了，然后又有人重启后的情况
				ActivemqHost ah = null;
				List<ActivemqHost>  rlt = new ArrayList<ActivemqHost>();
				for(Entry<String, String> e:map.entrySet()){
					ah = new ActivemqHost();
					ah.setName(e.getKey());
					ah.setIp(e.getValue());
					ah.setCurrentUse(e.getKey().equals(currentHostName));
					rlt.add(ah);							
				}
				
				return rlt;
			}else if(list.size() == 2){
				ActivemqHost ah = null;
				List<ActivemqHost> rlt = new ArrayList<ActivemqHost>();
				for(Entry<String, String> e:map.entrySet()){
					ah = new ActivemqHost();
					ah.setIp(e.getValue());
					ah.setName(e.getKey());
					ah.setCurrentUse(e.getKey().equals(currentHostName));
					if(previousHostIp != null){
						ah.setBroken(e.getValue().equals(previousHostIp));
					}else{
						ah.setBroken(e.getKey().equals(AcqclusterConfig.previousHostName));
					}
					rlt.add(ah);							
				}
				
				return rlt;
			}
		}
		return null;
	}
	
	private List<ActivemqHost> allTheSame(boolean allBroken,String currentHostName,Map<String,String> map){
		ActivemqHost ah = null;
		List<ActivemqHost> rlt = new ArrayList<ActivemqHost>();
		for(Entry<String, String> e:map.entrySet()){
			ah = new ActivemqHost();
			ah.setIp(e.getValue());
			ah.setName(e.getKey());
			if(allBroken){
				ah.setCurrentUse(false);
			}else{
				ah.setCurrentUse(e.getKey().equals(currentHostName));
			}
			ah.setBroken(allBroken);
			rlt.add(ah);							
		}
		return rlt;
	}
	
}
